/*
 *    Copyright 2012-2014 the original author or authors.
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */
package net.developerstation.mybatis.caches.redis;

import java.util.ArrayList;
import java.util.List;

import net.developerstation.mybatis.caches.redis.builder.RedisConfiguration;
import net.developerstation.mybatis.caches.redis.builder.RedisConfigurationBuilder;
import net.developerstation.mybatis.caches.redis.utils.SerializationUtils;

import org.apache.ibatis.logging.Log;
import org.apache.ibatis.logging.LogFactory;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

/**
 * a wrapper for jedis client
 * 
 * @author taox
 */
public class JedisClientWrapper {
	
	private final Log log = LogFactory.getLog(JedisClientWrapper.class);
	
	private final RedisConfiguration configuration;
	private static JedisPool pool;
	public JedisClientWrapper(){
		configuration = RedisConfigurationBuilder.getInstance().parseConfiguration();
        pool = new JedisPool(configuration,
    				configuration.getHost(), 
    				configuration.getPort(),
    				configuration.getTimeout(),
    				configuration.getPassword(),
    				configuration.getDatabase(), 
    				configuration.getClientName());
        if (log.isDebugEnabled()) {
            log.debug("Running new Jedis client using " + configuration);
        }
	}
	

	public Object getObject(Object key,String id) {
		Object obj = null;
		boolean broken = false;
		Jedis cache = pool.getResource();
		try {
			if (null == key)
				return null;
			byte[] b = cache.get(getKeyName(id,key).getBytes());
			if(b != null)
				obj = SerializationUtils.deserialize(b);
		} catch (Exception e) {
			log.error("Error occured when get data from redis cache", e);
			broken = true;
		} finally {
			this.returnResource(cache, broken);
		}
		return obj;
	}

	public void putObject(Object key, Object value,String id) {
		if (value == null)
			removeObject(key,id);
		else {
			boolean broken = false;
			Jedis cache = pool.getResource();
			try {
				cache.set(getKeyName(id,key).getBytes(), SerializationUtils.serialize(value));
			} catch (Exception e) {
				broken = true;
				throw new CacheException(e);
			} finally {
				this.returnResource(cache, broken);
			}
		}
	}
	
	public Object removeObject(Object key,String id) {
		boolean broken = false;
		Jedis cache = pool.getResource();
		try {
			cache.del(getKeyName(id,key));
			return null;
		} catch (Exception e) {
			broken = true;
			throw new CacheException(e);
		} finally {
			this.returnResource(cache, broken);
		}
    }
	
	public void removeObject(List keys,String id) throws CacheException {
		if(keys == null || keys.size() == 0)
			return ;
		boolean broken = false;
		Jedis cache = pool.getResource();
		try {
			String[] okeys = new String[keys.size()];
			for(int i=0;i<okeys.length;i++){
				okeys[i] = getKeyName(id,keys.get(i));
			}
			cache.del(okeys);
		} catch (Exception e) {
			broken = true;
			throw new CacheException(e);
		} finally {
			this.returnResource(cache, broken);
		}
	}
	
	
	public List<String> getKeys(String id) {
		Jedis cache = pool.getResource();
		boolean broken = false;
		try {
			List<String> keys = new ArrayList<String>();
			keys.addAll(cache.keys(id + ":*"));
			for(int i=0;i<keys.size();i++){
				keys.set(i, keys.get(i).substring(id.length() + 3));
			}
			return keys;
		} catch (Exception e) {
			broken = true;
			throw new CacheException(e);
		} finally {
			this.returnResource(cache, broken);
		}
	}
	
	
	public void clear(String id) {
		Jedis cache = pool.getResource();
		boolean broken = false;
		try {
			cache.del(id + ":*");
		} catch (Exception e) {
			broken = true;
			throw new CacheException(e);
		} finally {
			this.returnResource(cache, broken);
		}
	}
	
	
	/**
	 * release connection resource
	 * @param jedis
	 * @param isBrokenResource
	 */
    public  void returnResource(Jedis jedis,boolean isBrokenResource) {
    	if(null == jedis)
    		return;
        if(isBrokenResource){     
        	pool.returnBrokenResource(jedis);
            jedis = null;
        }
        else
        	pool.returnResource(jedis);
    }
    
	/**
	 * generate cache key
	 * @param key
	 * @return
	 */
	@SuppressWarnings("rawtypes")
	private String getKeyName(String id,Object key) {

		if(key instanceof Number)
			return id + ":I:" + key;
		else{
			Class keyClass = key.getClass();
			if(String.class.equals(keyClass) || StringBuffer.class.equals(keyClass) || StringBuilder.class.equals(keyClass))
				return id + ":S:" + key;
		}
		return id + ":O:" + key;
	}
	
}
